package io.vertx.kue.http

import io.vertx.core.*
import io.vertx.core.impl.VertxThread
import io.vertx.kue.Kue
import io.vertx.kue.queue.*
import io.vertx.core.json.DecodeException
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import io.vertx.core.logging.LoggerFactory
import io.vertx.ext.web.Router
import io.vertx.ext.web.RoutingContext
import io.vertx.ext.web.handler.BodyHandler
import io.vertx.ext.web.handler.StaticHandler
import io.vertx.ext.web.templ.JadeTemplateEngine
import io.vertx.kotlin.core.json.json
import io.vertx.kotlin.core.json.obj
import java.util.*
import java.util.concurrent.atomic.AtomicInteger
import java.util.function.Function

/**
 * The verticle serving Kue UI and REST API.
 *
 * @author Eric Zhao
 */
class KueHttpVerticle : AbstractVerticle() {

    private var kue: Kue? = null
    private var engine: JadeTemplateEngine? = null

    private fun newRouter(vertx: Vertx): Router {
        // create route
        val router = Router.router(vertx)
        router.route().handler(BodyHandler.create())
        // REST API routes
        router.get(KUE_API_JOB_SEARCH).handler({ this.apiSearchJob(it) })
        router.get(KUE_API_STATS).handler({ this.apiStats(it) })
        router.get(KUE_API_TYPE_STATE_STATS).handler({ this.apiTypeStateStats(it) })
        router.get(KUE_API_GET_JOB_TYPES).handler({ this.apiJobTypes(it) })
        router.get(KUE_API_JOB_RANGE).handler({ this.apiJobRange(it) }) // \/jobs\/([0-9]*)\.\.([0-9]*)(\/[^\/]+)?
        router.get(KUE_API_JOB_TYPE_RANGE).handler({ this.apiJobTypeRange(it) })
        router.get(KUE_API_JOB_STATE_RANGE).handler({ this.apiJobStateRange(it) })
        router.get(KUE_API_JOB_RANGE_ORDER).handler({ this.apiJobRange(it) })
        router.put(KUE_API_CREATE_JOB).handler({ this.apiCreateJob(it) })
        router.put(KUE_API_UPDATE_JOB_STATE).handler({ this.apiUpdateJobState(it) })
        router.get(KUE_API_GET_JOB).handler({ this.apiGetJob(it) })
        router.get(KUE_API_GET_JOB_LOG).handler({ this.apiFetchLog(it) })
        router.delete(KUE_API_DELETE_JOB).handler({ this.apiDeleteJob(it) })
        router.post(KUE_API_RESTART_JOB).handler({ this.apiRestartJob(it) })
        // UI routes
        router.route(KUE_UI_ROOT).handler({ this.handleUIRoot(it) })
        router.route(KUE_UI_ACTIVE).handler({ this.handleUIActive(it) })
        router.route(KUE_UI_INACTIVE).handler({ this.handleUIInactive(it) })
        router.route(KUE_UI_COMPLETE).handler({ this.handleUIComplete(it) })
        router.route(KUE_UI_FAILED).handler({ this.handleUIFailed(it) })
        router.route(KUE_UI_DELAYED).handler({ this.handleUIADelayed(it) })
        // static resources route
        router.route().handler(StaticHandler.create())
        return router
    }

    @Throws(Exception::class)
    override fun start(future: Future<Void>) {
        // init kue
        kue = Kue.createQueue(vertx, config())
        engine = JadeTemplateEngine.create()
        val router = newRouter(vertx)
        val size = 1
        val count = AtomicInteger(size)
        for (i in 1 .. size) {
            // create server
            vertx.createHttpServer()
                    .requestHandler {r -> router.accept(r) }
                    .listen(config().getInteger("http.port", PORT)!!,
                            config().getString("http.address", HOST)) { result ->
                        if (result.succeeded()) {
                            println("Kue http server is running on $PORT port...")
                            val i = count.decrementAndGet()
                            if (i == 0) {
                                future.complete()
                            }
                        } else {
                            future.fail(result.cause())
                        }
                    }
        }
    }

    /**
     * Render UI by job state
     *
     * @param state job state
     */
    private fun render(context: RoutingContext, state: String) { // TODO: bug in `types` param
        val uiPath = "webroot/views/job/list.jade"
        val title = config().getString("kue.ui.title", "Vert.x Kue")
        kue!!.allTypes
                .setHandler(resultHandler<List<String>>(context, Handler{ r ->
                    context.put("state", state)
                            .put("types", r)
                            .put("title", title)
                    engine!!.render(context, uiPath, { res ->
                        if (res.succeeded()) {
                            context.response()
                                    .putHeader("content-type", "text/html")
                                    .end(res.result())
                        } else {
                            context.fail(res.cause())
                        }
                    })
                }))
    }

    private fun handleUIRoot(context: RoutingContext) {
        handleUIActive(context) // by default active
    }

    private fun handleUIInactive(context: RoutingContext) {
        render(context, "inactive")
    }

    private fun handleUIFailed(context: RoutingContext) {
        render(context, "failed")
    }

    private fun handleUIComplete(context: RoutingContext) {
        render(context, "complete")
    }

    private fun handleUIActive(context: RoutingContext) {
        render(context, "active")
    }

    private fun handleUIADelayed(context: RoutingContext) {
        render(context, "delayed")
    }

    private fun apiSearchJob(context: RoutingContext) {
        notImplemented(context) // TODO: Not Implemented
    }

    private fun apiStats(context: RoutingContext) {
        val current = Thread.currentThread()
        when (current) {
            is VertxThread -> {
                val isworker = current.isWorker
                val n = current.name
                val ct = vertx.orCreateContext
                LOGGER.info("apiStats: [${current.id} -$n] ${ct.deploymentID()} ${ct}")
//                Thread.sleep(100)
                LOGGER.info("apiStats:End [${current.id} -$n]")
            }
        }

        val stats = JsonObject()
        kue!!.workTime.compose({ r ->
            stats.put("workTime", r)
            kue!!.getIdsByState(JobState.INACTIVE)
        }).compose({ r ->
            stats.put("inactiveCount", r.size)
            kue!!.getIdsByState(JobState.COMPLETE)
        }).compose({ r ->
            stats.put("completeCount", r.size)
            kue!!.getIdsByState(JobState.ACTIVE)
        }).compose({ r ->
            stats.put("activeCount", r.size)
            kue!!.getIdsByState(JobState.FAILED)
        }).compose({ r ->
            stats.put("failedCount", r.size)
            kue!!.getIdsByState(JobState.DELAYED)
        }).map({ r ->
            stats.put("delayedCount", r.size)
            stats
        }).setHandler(resultHandler<JsonObject>(context, Handler{ r ->
            context.response()
                    .putHeader("content-type", "application/json")
                    .end(r.encodePrettily())
        }))
    }

    private fun apiTypeStateStats(context: RoutingContext) {
        try {
            val type = context.request().getParam("type")
            val state = JobState.valueOf(context.request().getParam("state").toUpperCase())
            kue!!.cardByType(type, state).setHandler(resultHandler<Long>(context, Handler{ r ->
                context.response()
                        .putHeader("content-type", "application/json")
                        .end(JsonObject().put("count", r).encodePrettily())
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    private fun apiJobTypes(context: RoutingContext) {
        kue!!.allTypes.setHandler(resultHandler<List<String>>(context, Handler{ r ->
            context.response()
                    .putHeader("content-type", "application/json")
                    .end(JsonArray(r).encodePrettily())
        }))
    }

    private fun apiCreateJob(context: RoutingContext) {
        try {
            val job = Job(JsonObject(context.getBodyAsString())) // TODO: support json array create
            job.save().setHandler(resultHandler(context, Handler{ r ->
                val result = JsonObject().put("message", "job created")
                        .put("id", r.getId())
                        .encodePrettily()
                context.response().setStatusCode(201)
                        .putHeader("content-type", "application/json")
                        .end(result)
            }))
        } catch (e: DecodeException) {
            badRequest(context, e)
        }

    }

    private fun apiUpdateJobState(context: RoutingContext) {
        try {
            val id = java.lang.Long.parseLong(context.request().getParam("id"))
            val state = JobState.valueOf(context.request().getParam("state").toUpperCase())
            kue!!.getJob(id)
                    .compose({ j1 ->
                        if (j1.isPresent()) {
                            j1.get().state(state).compose(Job::save)
                        } else {
                            Future.succeededFuture<Job>()
                        }
                    })
                    .setHandler(resultHandler(context, Handler{ job ->
                if (job != null) {
                    context.response().putHeader("content-type", "application/json")
                            .end(JsonObject().put("message", "job_state_updated").encodePrettily())
                } else {
                    context.response().setStatusCode(404)
                            .putHeader("content-type", "application/json")
                            .end(JsonObject().put("message", "job_not_found").encodePrettily())
                }
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    private fun apiGetJob(context: RoutingContext) {
        try {
            val id = java.lang.Long.parseLong(context.request().getParam("id"))
            kue!!.getJob(id).setHandler(resultHandler<Optional<Job>>(context, Handler{ r ->
                if (r.isPresent()) {
                    context.response()
                            .putHeader("content-type", "application/json")
                            .end(r.get().toString())
                } else {
                    notFound(context)
                }
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    private fun apiJobRange(context: RoutingContext) {
        try {
            var order = context.request().getParam("order")
            if (order == null || !isOrderValid(order))
                order = "asc"
            val from = java.lang.Long.parseLong(context.request().getParam("from"))
            val to = java.lang.Long.parseLong(context.request().getParam("to"))
            kue!!.jobRange(from, to, order)
                    .setHandler(resultHandler<List<Job>>(context, Handler{ r ->
                        val result = JsonArray(r).encodePrettily()
                        context.response()
                                .putHeader("content-type", "application/json")
                                .end(result)
                    }))
        } catch (e: Exception) {
            e.printStackTrace()
            badRequest(context, e)
        }

    }

    private fun apiJobTypeRange(context: RoutingContext) {
        try {
            var order = context.request().getParam("order")
            if (order == null || !isOrderValid(order)) {
                order = "asc"
            }
            val from = java.lang.Long.parseLong(context.request().getParam("from"))
            val to = java.lang.Long.parseLong(context.request().getParam("to"))
            val state = context.request().getParam("state")
            val type = context.request().getParam("type")
            kue!!.jobRangeByType(type, state, from, to, order)
                    .setHandler(resultHandler<List<Job>>(context, Handler{ r ->
                        val result = JsonArray(r).encodePrettily()
                        context.response()
                                .putHeader("content-type", "application/json")
                                .end(result)
                    }))
        } catch (e: Exception) {
            e.printStackTrace()
            badRequest(context, e)
        }

    }

    private fun apiJobStateRange(context: RoutingContext) {
        try {
            var order = context.request().getParam("order")
            if (order == null || !isOrderValid(order))
                order = "asc"
            val from = java.lang.Long.parseLong(context.request().getParam("from"))
            val to = java.lang.Long.parseLong(context.request().getParam("to"))
            val state = context.request().getParam("state")
            kue!!.jobRangeByState(state, from, to, order)
                    .setHandler(resultHandler<List<Job>>(context, Handler{ r ->
                        val result = JsonArray(r).encodePrettily()
                        context.response()
                                .putHeader("content-type", "application/json")
                                .end(result)
                    }))
        } catch (e: Exception) {
            e.printStackTrace()
            badRequest(context, e)
        }

    }

    private fun isOrderValid(order: String): Boolean {
        return order == "asc" && order == "desc"
    }

    private fun apiDeleteJob(context: RoutingContext) {
        try {
            val id = java.lang.Long.parseLong(context.request().getParam("id"))
            kue!!.removeJob(id).setHandler(resultHandler<Void>(context, Handler{ r ->
                context.response().setStatusCode(204)
                        .putHeader("content-type", "application/json")
                        .end(JsonObject().put("message", "job $id removed").encodePrettily())
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    private fun apiRestartJob(context: RoutingContext) {
        try {
            val id = java.lang.Long.parseLong(context.request().getParam("id"))
            kue!!.getJob(id).setHandler(resultHandler<Optional<Job>>(context, Handler{ r ->
                if (r.isPresent()) {
                    r.get().inactive().setHandler(resultHandler(context, Handler{ r1 ->
                        context.response()
                                .putHeader("content-type", "application/json")
                                .end(JsonObject().put("message", "job $id restart").encodePrettily())
                    }))
                } else {
                    notFound(context)
                }
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    private fun apiFetchLog(context: RoutingContext) {
        try {
            val id = java.lang.Long.parseLong(context.request().getParam("id"))
            kue!!.getJobLog(id).setHandler(resultHandler<JsonArray>(context, Handler{ r ->
                context.response().putHeader("content-type", "application/json")
                        .end(r.encodePrettily())
            }))
        } catch (e: Exception) {
            badRequest(context, e)
        }

    }

    // helper methods

    /**
     * Wrap the result handler with failure handler (503 Service Unavailable)
     */
    private fun <T> resultHandler(context: RoutingContext, handler: Handler<T>): Handler<AsyncResult<T>> {
        return Handler{ res ->
            if (res.succeeded()) {
                handler.handle(res.result())
            } else {
                serviceUnavailable(context, res.cause())
            }
        }
    }

    private fun sendError(statusCode: Int, context: RoutingContext) {
        context.response().setStatusCode(statusCode).end()
    }

    private fun badRequest(context: RoutingContext, ex: Throwable) {
        context.response().setStatusCode(400)
                .putHeader("content-type", "application/json")
                .end(JsonObject().put("error", ex.message).encodePrettily())
    }

    private fun notFound(context: RoutingContext) {
        context.response().setStatusCode(404).end()
    }

    private fun notImplemented(context: RoutingContext) {
        context.response().setStatusCode(501)
                .end(JsonObject().put("message", "not_implemented").encodePrettily())
    }

    private fun serviceUnavailable(context: RoutingContext) {
        context.response().setStatusCode(503).end()
    }

    private fun serviceUnavailable(context: RoutingContext, ex: Throwable) {
        context.response().setStatusCode(503)
                .putHeader("content-type", "application/json")
                .end(JsonObject().put("error", ex.message).encodePrettily())
    }

    companion object {

        private val LOGGER = LoggerFactory.getLogger(KueHttpVerticle::class.java)

        private val HOST = "0.0.0.0"
        private val PORT = 8080 // Default port

        // Kue REST API
        private val KUE_API_JOB_SEARCH = "/job/search/:q"
        private val KUE_API_STATS = "/stats"
        private val KUE_API_TYPE_STATE_STATS = "/jobs/:type/:state/stats"
        private val KUE_API_GET_JOB = "/job/:id"
        private val KUE_API_GET_JOB_TYPES = "/job/types"
        private val KUE_API_JOB_RANGE = "/jobs/:from/to/:to"
        private val KUE_API_JOB_TYPE_RANGE = "/jobs/:type/:state/:from/to/:to/:order"
        private val KUE_API_JOB_STATE_RANGE = "/jobs/:state/:from/to/:to/:order"
        private val KUE_API_JOB_RANGE_ORDER = "/jobs/:from/to/:to/:order"
        private val KUE_API_CREATE_JOB = "/job"
        private val KUE_API_UPDATE_JOB_STATE = "/job/:id/state/:state"
        private val KUE_API_DELETE_JOB = "/job/:id"
        private val KUE_API_GET_JOB_LOG = "/job/:id/log"
        private val KUE_API_RESTART_JOB = "/inactive/:id"

        // Kue UI
        private val KUE_UI_ROOT = "/"
        private val KUE_UI_ACTIVE = "/active"
        private val KUE_UI_INACTIVE = "/inactive"
        private val KUE_UI_FAILED = "/failed"
        private val KUE_UI_COMPLETE = "/complete"
        private val KUE_UI_DELAYED = "/delayed"
    }
}

fun main(args: Array<String>) {
    val options = VertxOptions()
    options.setClustered(true)
    val vertx = Vertx.clusteredVertx(options, { res ->
        if (res.succeeded()) {
            val vertx = res.result()
            val doptions = DeploymentOptions()
            doptions.instances = 5
            vertx.deployVerticle(KueHttpVerticle::class.java, doptions)
        }
    })
}